package pageRank;

import java.io.IOException;
import java.util.Arrays;

import mapred.AbstractMapper;
import mapred.util.Bytes;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

public class Calc_Mapper extends AbstractMapper<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {

	@Override
	public void map(BytesWritable key, BytesWritable value,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		byte[] Z = new byte[16];
		Arrays.fill(Z, (byte)0);
		double z = Double.parseDouble(this.getJobConf().get("Z"));
		ReadBytes v = new ReadBytes(value);
		double pagerank = v.nextDouble();
		output.collect(key, new BytesWritable(Bytes.concat(new byte[]{0}, 
				Bytes.Double2Bytes(pagerank))));
		pagerank += z;
		byte[] t = v.nextBytes(v.remain());
		output.collect(key, new BytesWritable(t));
		if(t.length == 0) {
			output.collect(new BytesWritable(Z), 
					new BytesWritable(Bytes.Double2Bytes(pagerank)));
		} else {
			ReadBytes targets = new ReadBytes(new BytesWritable(t));
			while(targets.hasNextUrl()) {
				BytesWritable vv = new BytesWritable(Bytes.Double2Bytes(pagerank / (targets.size() / 16)));
				output.collect(new BytesWritable(targets.nextUrl()), vv);
			}
		}
	}

//	@Override
//	public void map(Text key, Text value, OutputCollector<Text, Text> output,
//			Reporter reporter) throws IOException {
//		double z = Double.parseDouble(this.getJobConf().get("Z"));
//		String str = value.toString();
//		int pos = str.indexOf("\t");
//		double pagerank = Double.parseDouble(str.substring(0, pos));
//		output.collect(key, new Text("@" + pagerank));
//		pagerank += z;
//		str = str.substring(pos + 1).trim();
//		output.collect(key, new Text(str));
//		if (str.isEmpty()) {
//			output.collect(new Text("Z"), new Text("#" + pagerank));
//		} else {
//			String[] targets = str.split("\t");
//			for (String target : targets) {
//				Text send = new Text("#" + (pagerank / targets.length));
//				output.collect(new Text(target), send);
//			}
//		}
//	}

}
